主题
Go 并发模式：把任务分开以后，怎样一起收尾
worker pool、fan-out、pipeline 经常一起出现在同一段代码里。它们不是互斥的架构选项，而是从不同角度描述任务怎么分配、结果怎么流动。真正需要先决定的是：最多同时做多少事，任务失败后谁停下，下游不再读时上游怎样退出。
这篇文章围绕一批整数求平方展开：一个生产者发任务，两个 worker 计算，一个收集者读取结果。随后在同一个生命周期框架里解释信号量、发布订阅和 errgroup。示例使用 Go 1.27.2，errgroup 固定为独立模块 golang.org/x/sync v0.16.0。[1][2]
这些模式分别管什么
| 模式 | 分配或连接关系 | 主要解决的问题 | 必须额外决定的事 |
|---|---|---|---|
| worker pool | 固定数量 worker 从任务队列取活 | 给活跃处理者设上限 | 队列容量、停止接单、排空还是丢弃 |
| fan-out | 一个输入分给多个处理者 | 让独立任务并行处理 | 单条输入通常只给一个接收者；结果顺序可能改变 |
| fan-in | 多个结果来源汇到一个出口 | 统一消费和结束通知 | 所有发送者结束后谁关闭输出 |
| pipeline | 多个阶段用 channel 连接 | 把输入、变换、输出组合起来 | 每一段背压、错误和提前退出如何传播 |
| 信号量 | 进入资源区前获取名额，离开时归还 | 限制并发资源占用 | 在哪里排队，获取失败后如何收尾 |
| 发布订阅 | 每个订阅者有自己的接收路径 | 同一事件送给多个接收者 | 慢订阅者阻塞、丢弃还是断开 |
| errgroup | 一组函数属于同一项工作 | 等待、首个错误与协同取消 | 下游是否配合取消，错误是否需要全部收集 |
这些名称表达设计结构，Go 并没有一个内建的 pipeline 类型。官方 pipeline 文章也是按阶段、channel 和 goroutine 的关系定义它们。[1:1]
一份完整的小流水线
正在绘制图示…
查看图源代码
flowchart LR
V[输入切片] --> P[一个生产者]
P --> J[jobs：无缓冲]
J --> W1[worker 1：求平方]
J --> W2[worker 2：求平方]
W1 --> R[results：无缓冲]
W2 --> R
R --> C[单一收集者]
W1 -. 结束计数 .-> X[协调者：等所有 worker]
W2 -. 结束计数 .-> X
X -. 最后关闭 .-> R
K[共同 context] -. 取消输入与输出等待 .-> P
K -. 取消 .-> W1
K -. 取消 .-> W2
同一个 jobs 被两个 worker 接收，是 fan-out：一个任务被一个 worker 取走，不会自动复制给两人。两个 worker 都发到 results，是共享输出形式的 fan-in；也可以让每个 worker 有独立输出，再写一个 merge 阶段。整个“输入、计算、收集”的连接又是一条 pipeline。[1:2][3]
文末 Squares 的调用很普通：
go
out, err := Squares(context.Background(), []int{1, 2, 3, 4}, 2)
fmt.Println(out, err)本次实际输出：
text
[1 4 9 16] <nil>这个稳定顺序是函数最后显式排序得到的。worker 完成顺序没有保证；如果业务要求与输入逐项对应，应该给每个任务附上下标，结果写回对应位置，而不是把结果按值排序。平方示例能排序，是因为这里仅要求结果集合。[4]
先确定关闭者，再启动工作
| 资源 | 谁写/发 | 谁关闭或等待 | 为什么没有发送撞上关闭 |
|---|---|---|---|
| jobs | 唯一生产者 | 生产者 defer close | 最后一次发送后才返回 |
| results | 所有 worker | 协调者等 producers.Wait 后关闭 | Wait 表明所有 worker 已不再发送 |
| 共同 context | 调用方可取消，Squares 也持有 cancel | Squares 返回前 cancel | 是退出请求，不负责关闭数据 channel |
| 全部自建 goroutine | 生产者、worker、关闭协调者 | all.Wait | 函数返回前把自己启动的工作等齐 |
代码使用两个 WaitGroup：producers 只数结果发送者，用来判断何时能安全关闭 results；all 数本函数启动的所有 goroutine，用来确认生命周期已收束。把两者混成一个计数，如果“关闭 results 的协调者”也在同一个它自己等待的计数里，就会等到自己，形成死锁。[5]
背压从接收者一路传回生产者
本例两个 channel 都无缓冲。收集者忙，worker 的发送会等待；worker 不能继续取下一个任务，生产者的发送也会等待。这条等待链把输入速度约束在下游能承受的范围内。
缓冲能吸收短时波动，但必须决定容量含义。队列里一个元素如果是指向大请求的指针，仅数“有 1000 个槽位”还看不出实际内存占用。固定 worker 数也不等于整个系统有界：如果任务来源先把百万个请求全读进切片，内存仍可增长；如果在每个请求里先启动 goroutine 再等名额，等待 goroutine 的数量仍可能无限。[1:3][3:1]
本例输入已经是一个有限切片，输出也累积在切片里，所以适合演示一批有限工作，并非无限流方案。无限或很大的输入通常需要逐条读取与逐条输出，而不是全量收集。
提前返回最容易漏掉的是发送端
错误路径经常长这样：消费者只需要第一个结果，拿到后就返回；剩下的 worker 还在向 results 发送，永远等不到接收者。给 results 增加一个任意大小的缓冲，只能在刚好装得下时掩盖问题。[1:4]
本例在生产者发送 jobs 和 worker 发送 results 的地方，都选择“发送成功，或者取消”：
go
select {
case results <- n * n:
case <-ctx.Done():
return
}接收任务时也监听同一个 ctx。这样父级取消后，等待输入和等待输出都有退出路径。Squares 的收集者会一直读到 results 关闭，再等 all；不会在仍有发送者存在时直接抛下输出。
| 发生什么 | 这个示例怎样收尾 | 需要知道的边界 |
|---|---|---|
| 输入处理完 | 关闭 jobs，worker 退出，协调者关 results，收集结束 | 无外部 I/O，平方计算会迅速返回 |
| 父级取消 | 每个 channel 阻塞点都能观察 ctx，最后等待所有 goroutine | 已拿到的任务可能继续算一次；select 不给取消优先权 |
| worker 数为 0 或负数 | 启动前直接返回错误 | 避免生产者没有接收者 |
| 输入为空 | 生产者直接关闭 jobs，后续自然结束 | 不要求特殊的 sleep 或轮询 |
本例最后检查 parent.Err()，所以父级已经取消时返回取消错误，丢弃部分结果。这是本函数选择的接口语义；业务如果要返回部分成功结果，要把“结果是否完整”一起表达，不能悄悄当作全量成功。
限并发与限速：先分清名额代表什么
容量为 N 的 channel 或 semaphore.Weighted 都能用作并发名额。一个任务进入前获取，完成时释放，相当于限制同时占着某种资源的工作数量。[6]
| 限制目标 | 控制量 | 适合的机制 | 一个常见误解 |
|---|---|---|---|
| 同时最多 20 个下游请求 | 当前在途数量 | 固定 worker、errgroup.SetLimit、信号量 | 20 并发并不等于每秒 20 次 |
| 同时最多占用 100 单位内存预算 | 活跃任务的权重总和 | 加权信号量 | 一个任务可能占多个名额 |
| 每秒平均 50 次、允许少量突发 | 随时间补充的令牌 | 令牌桶，如 x/time/rate | 队列容量不会自动变成速率限制 |
如果每个任务只需一毫秒，20 并发的完成速率可能很高；每个任务要一秒时又完全不同。因此不要把 semaphore 命名为“每秒速率限制器”。x/time/rate 是另一个模块，本篇只给出机制边界，没有安装或测试该包。[7]
获取名额最好放在启动 goroutine 之前，这样提交者本身承担背压。已经成功获取后要配对释放；获取被取消而失败则不能 Release。实际工作也要使用 ctx，信号量只让“等名额”可取消，不能结束已经进入临界资源区的任务。加权版需先校验 0 < weight <= limit；文末的工具选型专文有完整加权实现。这里不是依赖链接替代原理：关键就是让所有活跃任务的权重之和不超过预算。[6:1]
发布订阅：同一事件要送多份，慢订阅者要有政策
多个 worker 读同一个 channel 会分摊任务，不会各得一份。要广播，发布者需要向每个订阅者自己的队列发送。于是一个新的问题出现了：其中一个人一直不读，要不要让其他人也收不到后面的事件？
| 慢订阅者策略 | 好处 | 代价与适用边界 |
|---|---|---|
| 阻塞等待 | 在进程正常运行且消费持续推进的前提下，不主动丢事件 | 一个慢订阅者可拖住整个发布循环 |
| 缓冲满就丢本次事件 | 发布者和其他订阅者继续前进 | 接收方必须能容忍缺失，如状态刷新通知 |
| 断开慢订阅者 | 避免长期占住资源 | 需要重连、重放或重新拉取快照 |
| 写入持久消息系统 | 能进一步设计确认、重试、恢复 | 增加服务、存储与一致性成本；不再只是内存 channel |
下面实现选择“丢本次事件”，并把丢弃数量返回给调用者：
go
func Publish(subs []chan int, value int) (dropped int) {
for _, sub := range subs {
select {
case sub <- value:
default:
dropped++
}
}
return
}两个订阅队列容量都为 1。第一次发布 1 后，只让 fast 取走它，再发布 2。本次输出：
text
first dropped: 0
fast: 1
second dropped: 1
remaining: 2 1slow 还装着 1，因此第二条丢给了 slow，fast 收到了 2。这里没有“保证送达”的暗示，输出直接把丢弃结果暴露出来。该教学 broker 由一个调用方独占管理订阅列表、发布和关闭；不能让另一个 goroutine 随意关闭其中的 channel，否则 select 也挡不住发送到已关闭 channel 的 panic。增加并发订阅/退订时，需要让同一个 broker 循环拥有这些动作，或另设完整同步协议。[8][4:1]
errgroup：一个分支失败，让其余分支知道
WaitGroup 能等齐，却不收错误。errgroup 把一组返回 error 的函数放在一起，Wait 等所有函数返回，再给出第一个非 nil 错误；WithContext 返回的派生 ctx 在首个错误出现时取消，也在 Wait 返回时取消。[2:1]
go
g, ctx := errgroup.WithContext(context.Background())
started := make(chan struct{})
g.Go(func() error { close(started); <-ctx.Done(); return ctx.Err() })
g.Go(func() error { <-started; return errors.New("stage failed") })
fmt.Println("errgroup:", g.Wait())本次输出：
text
errgroup: stage failed第一个分支先开始等待，第二个分支报业务错误，于是第一个收到取消并返回。errgroup 保存的是触发取消的那个错误，而不是随后产生的 canceled。它不会把所有错误自动 Join，也不会把 panic 自动变成 error。[2:2]
把 errgroup 放进 pipeline 时要小心等待环：如果 worker 的退出依赖消费者继续读，而消费者在开始读之前先 g.Wait()，双方仍然会卡住。错误聚合工具不会替你改变数据流。输出的关闭协调也必须等“所有输出发送者”结束，而不能提前关闭。
如果只需要处理一个有限 slice，可用 errgroup.SetLimit 限活跃任务，少写一个任务队列。若输入是连续流、需要独立的排队容量、优先级或拒绝策略，显式 worker pool 往往更清楚。SetLimit 会让提交的 Go 调用等待名额，这次等待并没有 context 参数；任务还在运行时也不能修改 limit。复杂入队取消需求可选择固定 worker 或带 ctx 的 semaphore.Acquire。[2:3]
依据与本次验证
本次在 Go 1.27.2、Linux/amd64 实际运行有限 pipeline、丢弃型发布订阅和 errgroup 失败取消。测试覆盖正常计算、预先取消、空输入、非法 worker 数、慢订阅者丢弃及分支失败；go test -race -count=1 ./...、go vet ./... 通过。没有对无限输入、真实网络或持久消息系统做实验，也没有把平方任务的吞吐代表业务负载。
完整代码与复现方法
以同一目录为根，按下面标出的相对路径保存文件。使用 Go 1.27.2 运行：
sh
go run .
go test -race -count=1 ./...
go vet ./...依赖固定为 golang.org/x/sync v0.16.0。首次运行需要该版本已在模块缓存中，或能够访问 Go 模块代理；它是独立模块，不随 Go 1.27.2 自动更新。
go.mod
go.mod
module example.com/go-articles/05_patterns
go 1.25.0
require golang.org/x/sync v0.16.0go.sum
text
golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=main.go
代码块 · 101 行 · 点击收起 / 展开
go
package main
import (
"context"
"errors"
"fmt"
"golang.org/x/sync/errgroup"
"slices"
"sync"
)
// Squares owns every goroutine it starts and returns only after all exit.
func Squares(parent context.Context, values []int, workers int) ([]int, error) {
if workers <= 0 {
return nil, errors.New("workers must be positive")
}
ctx, cancel := context.WithCancel(parent)
defer cancel()
jobs := make(chan int)
results := make(chan int)
var all, producers sync.WaitGroup
all.Add(1)
go func() {
defer all.Done()
defer close(jobs)
for _, n := range values {
select {
case jobs <- n:
case <-ctx.Done():
return
}
}
}()
for range workers {
producers.Add(1)
all.Add(1)
go func() {
defer producers.Done()
defer all.Done()
for {
select {
case <-ctx.Done():
return
case n, ok := <-jobs:
if !ok {
return
}
select {
case results <- n * n:
case <-ctx.Done():
return
}
}
}
}()
}
all.Add(1)
go func() { defer all.Done(); producers.Wait(); close(results) }()
var out []int
for n := range results {
out = append(out, n)
}
all.Wait()
if err := parent.Err(); err != nil {
return nil, err
}
slices.Sort(out)
return out, nil
}
// The broker is single-owner: only its caller touches subscribers and closes them.
func Publish(subs []chan int, value int) (dropped int) {
for _, sub := range subs {
select {
case sub <- value:
default:
dropped++
}
}
return
}
func CancelPeers() error {
g, ctx := errgroup.WithContext(context.Background())
started := make(chan struct{})
g.Go(func() error { close(started); <-ctx.Done(); return ctx.Err() })
g.Go(func() error { <-started; return errors.New("stage failed") })
return g.Wait()
}
func main() {
out, err := Squares(context.Background(), []int{1, 2, 3, 4}, 2)
fmt.Println(out, err)
fast, slow := make(chan int, 1), make(chan int, 1)
subs := []chan int{fast, slow}
fmt.Println("first dropped:", Publish(subs, 1))
fmt.Println("fast:", <-fast)
fmt.Println("second dropped:", Publish(subs, 2))
close(fast)
close(slow)
fmt.Println("remaining:", <-fast, <-slow)
fmt.Println("errgroup:", CancelPeers())
}main_test.go
代码块 · 52 行 · 点击收起 / 展开
go
package main
import (
"context"
"errors"
"slices"
"testing"
)
func TestSquares(t *testing.T) {
got, err := Squares(context.Background(), []int{1, 2, 3, 4}, 2)
if err != nil || !slices.Equal(got, []int{1, 4, 9, 16}) {
t.Fatal(got, err)
}
}
func TestAlreadyCanceled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel()
_, err := Squares(ctx, make([]int, 100), 2)
if !errors.Is(err, context.Canceled) {
t.Fatal(err)
}
}
func TestEmptyAndInvalid(t *testing.T) {
if out, err := Squares(context.Background(), nil, 3); err != nil || len(out) != 0 {
t.Fatal(out, err)
}
if _, err := Squares(context.Background(), nil, 0); err == nil {
t.Fatal("expected error")
}
}
func TestPublishDropPolicy(t *testing.T) {
a, b := make(chan int, 1), make(chan int, 1)
if n := Publish([]chan int{a, b}, 7); n != 0 {
t.Fatal(n)
}
<-a
if n := Publish([]chan int{a, b}, 9); n != 1 {
t.Fatal(n)
}
if x, y := <-a, <-b; x != 9 || y != 7 {
t.Fatal(x, y)
}
close(a)
close(b)
}
func TestCancelPeers(t *testing.T) {
if e := CancelPeers(); e == nil || e.Error() != "stage failed" {
t.Fatal(e)
}
}Go 官方：Pipelines and cancellation，阶段定义、fan-out/fan-in、关闭与提前结束。本文使用自己的有限批处理和收尾接口。 ↩︎ ↩︎ ↩︎ ↩︎ ↩︎
x/sync v0.16.0 errgroup，
WithContext、Wait、Go、SetLimit。本次也读取了该缓存版本的完整 go doc。 ↩︎ ↩︎ ↩︎ ↩︎x/sync v0.16.0 semaphore，
Acquire、Release和权重限制。固定版本源码已在本次环境核对。 ↩︎ ↩︎golang.org/x/time/rate，令牌桶和 rate/burst 概念，仅用于与并发上限区别。 ↩︎